package com.enhanced.mqtt.manager;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.function.Consumer;

/**
 * EMQX MQTT服务器管理类
 * 集成了MQTT的高级特性，如遗嘱消息、保留消息、共享订阅等
 */
public class EmqxServerManager {
    // MQTT服务器相关配置
    private final String brokerUrl;
    private final String clientId;
    private final MqttAsyncClient client;
    
    // 消息处理回调
    private Consumer<MqttMessage> messageHandler;
    
    /**
     * 构造函数
     * @param brokerUrl MQTT服务器地址
     * @param clientId 客户端ID
     * @param persistence 持久化
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager(String brokerUrl, String clientId, MqttClientPersistence persistence) throws MqttException {
        this.brokerUrl = brokerUrl;
        this.clientId = clientId;
        this.client = new MqttAsyncClient(brokerUrl, clientId, persistence);
    }
    
    /**
     * 创建默认配置的管理器
     * @param brokerUrl MQTT服务器地址
     * @return 服务器管理器实例
     * @throws MqttException MQTT异常
     */
    public static EmqxServerManager createDefault(String brokerUrl) throws MqttException {
        String randomClientId = "JavaClient-" + UUID.randomUUID().toString().substring(0, 8);
        return new EmqxServerManager(brokerUrl, randomClientId, new MemoryPersistence());
    }
    
    /**
     * 设置消息处理回调
     * @param messageHandler 消息处理器
     * @return 当前管理器实例
     */
    public EmqxServerManager setMessageHandler(Consumer<MqttMessage> messageHandler) {
        this.messageHandler = messageHandler;
        return this;
    }
    
    /**
     * 连接到MQTT服务器
     * @param username 用户名
     * @param password 密码
     * @param willTopic 遗嘱主题
     * @param willMessage 遗嘱消息
     * @param cleanSession 清除会话
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager connect(String username, String password, 
                                    String willTopic, String willMessage,
                                    boolean cleanSession) throws MqttException {
        MqttConnectOptions options = new MqttConnectOptions();
        
        // 设置认证信息
        if (username != null && password != null) {
            options.setUserName(username);
            options.setPassword(password.toCharArray());
        }
        
        // 设置遗嘱消息
        if (willTopic != null && willMessage != null) {
            options.setWill(willTopic, willMessage.getBytes(), 1, false);
        }
        
        // 设置会话清除
        options.setCleanSession(cleanSession);
        
        // 设置心跳
        options.setKeepAliveInterval(60);
        
        // 设置自动重连
        options.setAutomaticReconnect(true);
        
        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("连接丢失: " + cause.getMessage());
                cause.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                // 如果设置了消息处理器，则调用
                if (messageHandler != null) {
                    messageHandler.accept(message);
                }
                
                // 默认打印消息内容
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                System.out.println("\n收到消息:");
                System.out.println("  主题: " + topic);
                System.out.println("  内容: " + payload);
                System.out.println("  QoS: " + message.getQos());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("消息发送完成");
            }
        });
        
        // 连接服务器
        System.out.println("正在连接到MQTT服务器: " + brokerUrl);
        client.connect(options).waitForCompletion();
        System.out.println("已连接到MQTT服务器，客户端ID: " + clientId);
        
        return this;
    }
    
    /**
     * 使用简单配置连接
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager connectWithDefaultOptions() throws MqttException {
        return connect(null, null, null, null, true);
    }
    
    /**
     * 订阅主题
     * @param topic 主题
     * @param qos 服务质量
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager subscribe(String topic, int qos) throws MqttException {
        client.subscribe(topic, qos).waitForCompletion();
        System.out.println("已订阅主题: " + topic + ", QoS: " + qos);
        return this;
    }
    
    /**
     * 使用共享订阅
     * @param groupName 组名
     * @param topic 主题
     * @param qos 服务质量
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager subscribeShared(String groupName, String topic, int qos) throws MqttException {
        String sharedTopic = "$share/" + groupName + "/" + topic;
        client.subscribe(sharedTopic, qos).waitForCompletion();
        System.out.println("已订阅共享主题: " + sharedTopic + ", QoS: " + qos);
        return this;
    }
    
    /**
     * 发布消息
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量
     * @param retained 是否保留
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager publish(String topic, String payload, int qos, boolean retained) throws MqttException {
        MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
        message.setQos(qos);
        message.setRetained(retained);
        
        client.publish(topic, message).waitForCompletion();
        System.out.println("已发布消息到主题: " + topic);
        return this;
    }
    
    /**
     * 发布延迟消息
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量
     * @param delaySeconds 延迟秒数
     * @return 当前管理器实例
     * @throws MqttException MQTT异常
     */
    public EmqxServerManager publishDelayed(String topic, String payload, int qos, int delaySeconds) throws MqttException {
        String delayedTopic = "$delayed/" + delaySeconds + "/" + topic;
        return publish(delayedTopic, payload, qos, false);
    }
    
    /**
     * 断开连接
     * @throws MqttException MQTT异常
     */
    public void disconnect() throws MqttException {
        if (client.isConnected()) {
            client.disconnect().waitForCompletion();
            System.out.println("已断开与MQTT服务器的连接");
        }
    }
    
    /**
     * 关闭客户端
     * @throws MqttException MQTT异常
     */
    public void close() throws MqttException {
        disconnect();
        client.close();
        System.out.println("客户端已关闭");
    }
    
    /**
     * 获取MQTT客户端
     * @return MQTT客户端
     */
    public MqttAsyncClient getClient() {
        return client;
    }
} 